package com.example.one;

import cn.hutool.core.collection.CollUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * TODO
 *
 * @author coxx_
 * @date 2020/10/30 10:49
 */
public class Job {
    public static void main(String[] args) throws Exception {
        // 设置保存位置
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        env.enableCheckpointing(10000);
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner()
                .inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, environmentSettings);
        DataStream<Order> orderA = env.fromCollection(CollUtil.toList(
                new Order("beer", 3.0, 1),
                new Order("diaper", 4.0, 1),
                new Order("rubber", 2.0, 3)));
        DataStream<Order> orderB = env.fromCollection(CollUtil.toList(
                new Order("pen", 3.0, 2),
                new Order("rubber", 3.0, 2),
                new Order("beer", 1.0, 4)));
        tEnv.createTemporaryView("OrderA", orderA);
        tEnv.createTemporaryView("OrderB", orderB);
        // 通过 sql 进行筛选
        String sql = "select * from (select * from OrderA where nums < 3 " +
                "union all select * from OrderB where name <> 'rubber') OrderAll where amount > 2";
        // catalog.database.table_name
        Table orderAll = tEnv.sqlQuery(sql);
        tEnv.toRetractStream(orderAll, Order.class).print();
        env.execute();
    }
}
